package com.event.ingestion.data.hbase;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

public class EventParser extends com.event.ingestion.data.EventParser<Put> {
    @Override
    public Put parse(String[] fields) {
        Put p = new Put(Bytes.toBytes(fields[0]));

        //schedule:start_time
        p = p.addColumn(Bytes.toBytes("schedule"),Bytes.toBytes("start_time"),Bytes.toBytes(fields[2]));

        //location:city
        p = p.addColumn(Bytes.toBytes("location"),Bytes.toBytes("city"),Bytes.toBytes(fields[3]));
        p = p.addColumn(Bytes.toBytes("location"),Bytes.toBytes("state"),Bytes.toBytes(fields[4]));
        p = p.addColumn(Bytes.toBytes("location"),Bytes.toBytes("zip"),Bytes.toBytes(fields[5]));
        p = p.addColumn(Bytes.toBytes("location"),Bytes.toBytes("country"),Bytes.toBytes(fields[6]));
        p = p.addColumn(Bytes.toBytes("location"),Bytes.toBytes("lat"),Bytes.toBytes(fields[7]));
        p = p.addColumn(Bytes.toBytes("location"),Bytes.toBytes("lng"),Bytes.toBytes(fields[8]));

        //creator
        p = p.addColumn(Bytes.toBytes("creator"),Bytes.toBytes("user_id"),Bytes.toBytes(fields[1]));

        StringBuilder stringBuilder = new StringBuilder();
        if (fields.length > 8) {
            for (int i = 9; i < fields.length; i++) {
                if (stringBuilder.length() > 0) {
                    stringBuilder.append("|");
                }
                stringBuilder.append(fields[i]);
            }
        }
        //remark
        p.addColumn(Bytes.toBytes("remark"),Bytes.toBytes("common_words"),Bytes.toBytes(stringBuilder.toString()));

        return p;
    }
}
